package cn.enjoyedu.rebalance;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author King老师   享学课堂 https://enjoy.ke.qq.com
 * 往期课程咨询芊芊老师  QQ：2130753077 VIP课程咨询 依娜老师  QQ：2133576719
 * 类说明：再均衡监听器
 */
public class HandlerRebalance implements ConsumerRebalanceListener {

    /*模拟一个保存分区偏移量的数据库表*/
    public final static ConcurrentHashMap<TopicPartition,Long> partitionOffsetDao = new ConcurrentHashMap<>();

    //MARK 消费者 消费时会同时更新这里面的偏移量
    private final Map<TopicPartition, OffsetAndMetadata> currOffsets;
    
    //MARK 本监听器 存在目的 就是为了这个消费者, 可以在发生再均衡的时候 也可以准确从上次结束的地方开始消化
    private final KafkaConsumer<String,String> consumer;
    //private final Transaction  tr事务类的实例

    public HandlerRebalance(Map<TopicPartition, OffsetAndMetadata> currOffsets,
                            KafkaConsumer<String, String> consumer) {
        this.currOffsets = currOffsets;
        this.consumer = consumer;
    }
    
    /**
     * 
     *  分区再均衡之前 会被调用
     *  
     *  其实一开始 消费者启动 就会这里调一次 但是这时候partitions还是空集合[]
     *  
     *  这里面我只是打印看看
     * @param partitions 本监听所属的消费者 的所有分区
     */
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        
        final String id = Thread.currentThread().getId()+"";//一个消费者一个线程, 这里是用线程号标识消费者
        System.out.println(id+"-onPartitionsRevoked参数值为："+partitions);
        System.out.println(id+"-服务器准备分区再均衡，提交偏移量。当前偏移量为："
                +currOffsets);
        

    }

    /**
     * 
     * 分区再均衡完成以后 这里会调用, 这里之后才开始消费
     * 这里消费者 一启动 就会被调用一次 , 这时候 partitions已经分配好了
     * @param partitions 本监听器的消费者 所有的分区
     */
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        
        final String id = Thread.currentThread().getId()+"";
        System.out.println(id+"-再均衡完成，onPartitionsAssigned参数值为："+partitions);
        System.out.println("分区偏移量表中："+ partitionOffsetDao);
        
        for(TopicPartition topicPartition:partitions){
            
            System.out.println(id+"-topicPartition"+topicPartition);
            
            Long offset = partitionOffsetDao.get(topicPartition);
            if(offset==null) {
                continue;
            }
            //MARK 设消费者的偏移量(自己在数据库里面存的 不靠kafka记),消费者从这开始消费 
            consumer.seek(topicPartition, offset);
        }

    }
}
